package com.shujia.mr.kqzl2;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

//reduce输入: <"1001",["$20180603-50","#北京-万寿西宫"]>   <"3033",["$20180603-50"]>
//reduce输出: <"20180603-北京-万寿西宫","50">
public class PM25JoinCityReducer extends Reducer<Text, Text, Text, Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {

        String city = "未知";
        String date = null;
        String avg_pm25 = null;

        for (Text value : values) {
            String info = value.toString();
            if (info.startsWith("$")) {
                //$20180603-50
                String[] strings = info.split("-");
                date = strings[0].substring(1);
                avg_pm25 = strings[1];
            } else if (info.startsWith("#")) {
                //#北京-万寿西宫
                city = info.substring(1);
            }
        }
        context.write(new Text(date + "-" + city), new Text(avg_pm25));
    }
}
